package ru.zombator.taskman.server.query.tcp;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

import ru.zombator.taskman.core.query.IPacket;
import ru.zombator.taskman.core.utils.PacketUtils;
import ru.zombator.taskman.server.query.BaseServer;

/**
 * TCP-сервер. Обрабатывает каждое новое соединение в отдельном потоке.
 * @author nvamelichev
 *
 */
public class TcpServer extends BaseServer {
	private static final long serialVersionUID = 1L;
	
	private final int TCP_SERVER_PORT = 3333;
	
	private ServerSocketChannel srvSocketChannel;

	private final ExecutorService responders = Executors.newCachedThreadPool();
	private final ExecutorService listener = Executors.newFixedThreadPool(1);
	private AtomicBoolean isRunning = new AtomicBoolean();
	
	private class ListenerThread implements Callable<Void> {
		@Override
		public Void call() {
			// ждем соединения
			while (isRunning.get()) {
				try {
					SocketChannel targetChannel = srvSocketChannel.accept();
					
					ResponderThread resp = new ResponderThread();
					resp.setTarget(targetChannel);
					responders.submit(resp);
				} catch (ClosedChannelException e) {
					// нас грубо прерывают методом close()
					return null;
				} catch (IOException e) {
				}
			}
			try {
				srvSocketChannel.close();
			} catch (IOException e) {
				e.printStackTrace();
			}			
			responders.shutdownNow();
			return null;
		}
	}
	
	private class ResponderThread implements Callable<Void> {
		private SocketChannel targetSocketChannel;
		
		public void setTarget(SocketChannel target) {
			this.targetSocketChannel = target;
		}
		
		@Override
		public Void call() throws Exception {
			targetSocketChannel.configureBlocking(true);
			
			ByteBuffer sizeBuf = ByteBuffer.allocate(4);
			sizeBuf.order(ByteOrder.LITTLE_ENDIAN);
			int readBytes = targetSocketChannel.read(sizeBuf);
			if (readBytes == -1)
				throw new IOException("unexpected disconnection");
			sizeBuf.rewind();
			int payloadSize = sizeBuf.getInt();
			
			ByteBuffer payloadBuf = ByteBuffer.allocate(payloadSize);
			payloadBuf.order(ByteOrder.LITTLE_ENDIAN);
			int payloadBytesRead = targetSocketChannel.read(payloadBuf);
			if (payloadBytesRead == -1)
				throw new IOException("unexpected disconnection");
			payloadBuf.rewind();
			IPacket request = PacketUtils.bufferToPacket(payloadBuf, payloadSize);
			
			String ipAddress = targetSocketChannel.socket().getInetAddress().getHostAddress();
			IPacket pktResponse = getResponse(request, ipAddress);
			ByteBuffer outBuf = PacketUtils.packetToBuffer(pktResponse);
			
			while (outBuf.hasRemaining()) {
				targetSocketChannel.write(outBuf);
			}
			
			targetSocketChannel.close();
			
			return null;
		}
	}
	
	public TcpServer() throws IOException {
		srvSocketChannel = ServerSocketChannel.open();
		srvSocketChannel.socket().bind(new InetSocketAddress(TCP_SERVER_PORT));
		srvSocketChannel.configureBlocking(true);
		
		isRunning.set(true);
		listener.submit(new ListenerThread());
	}
	
	@Override
	public void dispose() {
		isRunning.set(false);
		try {
			srvSocketChannel.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		listener.shutdown();
		responders.shutdown();
	}
}
